package com.sparrow.common.timer;

import com.sparrow.common.thread.factory.SimpleThreadFactory;
import io.netty.util.*;
import io.netty.util.internal.PlatformDependent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Collections;
import java.util.HashSet;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicLong;
import static io.netty.util.internal.StringUtil.simpleClassName;

/**
 * 基于 io.netty.util.HashedWheelTimer 二次修改的游戏服务器时间轮定时器
 * 每个Timer都会创建一条线程，一个应用只需创建一个Timer 共享即可。
 */
public class GameWheelTimer implements Timer {

    static final Logger logger = LoggerFactory.getLogger(GameWheelTimer.class);

    private static final AtomicInteger INSTANCE_COUNTER = new AtomicInteger();
    private static final AtomicBoolean WARNED_TOO_MANY_INSTANCES = new AtomicBoolean();
    private static final int INSTANCE_COUNT_LIMIT = 64;
    private static final ResourceLeakDetector<GameWheelTimer> leakDetector = ResourceLeakDetectorFactory.instance()
            .newResourceLeakDetector(GameWheelTimer.class, 1);

    private static final AtomicIntegerFieldUpdater<GameWheelTimer> WORKER_STATE_UPDATER =
            AtomicIntegerFieldUpdater.newUpdater(GameWheelTimer.class, "workerState");

    private final ResourceLeakTracker<GameWheelTimer> leak;
    private final GameWheelTimer.Worker worker = new GameWheelTimer.Worker();
    private final Thread workerThread;
    /**
     * 初始状态
     */
    public static final int WORKER_STATE_INIT = 0;
    /**
     * 已启动状态
     */
    public static final int WORKER_STATE_STARTED = 1;
    /**
     * 关闭状态
     */
    public static final int WORKER_STATE_SHUTDOWN = 2;
    @SuppressWarnings({ "unused", "FieldMayBeFinal" })
    private volatile int workerState; // 0 - 初始化, 1 - 已启动, 2 - 已关闭

    private final long tickDuration; // 每一滴答的时间间隔
    private final GameWheelTimer.HashedWheelBucket[] wheel; // 时间轮槽
    private final int mask; // 用于快速计算槽索引
    private final CountDownLatch startTimeInitialized = new CountDownLatch(1); // 启动时间初始化同步器
    private final Queue<GameWheelTimer.HashedWheelTimeout> timeouts = PlatformDependent.newMpscQueue(); // 待处理任务队列, newTimeOut 一开始都会先放这
    private final Queue<GameWheelTimer.HashedWheelTimeout> cancelledTimeouts = PlatformDependent.newMpscQueue(); // 被取消任务队列
    private final AtomicLong pendingTimeouts = new AtomicLong(0); // 待处理任务数量
    private final long maxPendingTimeouts; // 最大待处理任务数量

    private volatile long startTime; // 时间轮启动时间

    // ========================== 构造方法 ==========================
    public GameWheelTimer() {
        this(new SimpleThreadFactory("GameWheelTimer"));
    }

    public GameWheelTimer(long tickDuration, TimeUnit unit) {
        this(Executors.defaultThreadFactory(), tickDuration, unit);
    }

    public GameWheelTimer(long tickDuration, TimeUnit unit, int ticksPerWheel) {
        this(Executors.defaultThreadFactory(), tickDuration, unit, ticksPerWheel);
    }

    public GameWheelTimer(ThreadFactory threadFactory) {
        this(threadFactory, 100, TimeUnit.MILLISECONDS);
    }

    public GameWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit) {
        this(threadFactory, tickDuration, unit, 512);
    }

    public GameWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel) {
        this(threadFactory, tickDuration, unit, ticksPerWheel, true);
    }

    public GameWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection) {
        this(threadFactory, tickDuration, unit, ticksPerWheel, leakDetection, -1);
    }

    public GameWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel, boolean leakDetection, long maxPendingTimeouts) {

        if (threadFactory == null) {
            throw new NullPointerException("threadFactory");
        }
        if (unit == null) {
            throw new NullPointerException("unit");
        }
        if (tickDuration <= 0) {
            throw new IllegalArgumentException("tickDuration 必须大于 0: " + tickDuration);
        }
        if (ticksPerWheel <= 0) {
            throw new IllegalArgumentException("ticksPerWheel 必须大于 0: " + ticksPerWheel);
        }

        // 将 ticksPerWheel 规范化为 2 的幂，并初始化时间轮槽  ticksPerWheel 时间轮 槽位数量
        wheel = createWheel(ticksPerWheel);
        mask = wheel.length - 1;

        // 将 tickDuration 转换为纳秒
        this.tickDuration = unit.toNanos(tickDuration);

        // 防止溢出
        if (this.tickDuration >= Long.MAX_VALUE / wheel.length) {
            throw new IllegalArgumentException(String.format(
                    "tickDuration: %d (期望: 0 < tickDuration 纳秒 < %d",
                    tickDuration, Long.MAX_VALUE / wheel.length));
        }
        workerThread = threadFactory.newThread(worker);

        leak = leakDetection || !workerThread.isDaemon() ? leakDetector.track(this) : null;

        this.maxPendingTimeouts = maxPendingTimeouts;

        if (INSTANCE_COUNTER.incrementAndGet() > INSTANCE_COUNT_LIMIT &&
                WARNED_TOO_MANY_INSTANCES.compareAndSet(false, true)) {
            reportTooManyInstances();
        }
    }

    @Override
    protected void finalize() throws Throwable {
        try {
            super.finalize();
        } finally {
            // 对象即将被 GC，如果没有关闭则减少实例计数
            if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
                INSTANCE_COUNTER.decrementAndGet();
            }
        }
    }

    // ========================== 工具方法 ==========================
    private static GameWheelTimer.HashedWheelBucket[] createWheel(int ticksPerWheel) {
        if (ticksPerWheel <= 0) {
            throw new IllegalArgumentException("ticksPerWheel 必须大于 0: " + ticksPerWheel);
        }
        if (ticksPerWheel > 1073741824) {
            throw new IllegalArgumentException("ticksPerWheel 不得大于 2^30: " + ticksPerWheel);
        }

        ticksPerWheel = normalizeTicksPerWheel(ticksPerWheel);
        GameWheelTimer.HashedWheelBucket[] wheel = new GameWheelTimer.HashedWheelBucket[ticksPerWheel];
        for (int i = 0; i < wheel.length; i ++) {
            wheel[i] = new GameWheelTimer.HashedWheelBucket();
        }
        return wheel;
    }

    private static int normalizeTicksPerWheel(int ticksPerWheel) {
        int normalizedTicksPerWheel = 1;
        while (normalizedTicksPerWheel < ticksPerWheel) {
            normalizedTicksPerWheel <<= 1;
        }
        return normalizedTicksPerWheel;
    }

    // ========================== 启动 / 停止 ==========================
    public void start() {
        switch (WORKER_STATE_UPDATER.get(this)) {
            case WORKER_STATE_INIT:
                if (WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_INIT, WORKER_STATE_STARTED)) {
                    workerThread.start();
                }
                break;
            case WORKER_STATE_STARTED:
                break;
            case WORKER_STATE_SHUTDOWN:
                throw new IllegalStateException("无法在停止后再次启动");
            default:
                throw new Error("无效的 WorkerState");
        }

        // 等待 worker 初始化 startTime
        while (startTime == 0) {
            try {
                startTimeInitialized.await();
            } catch (InterruptedException ignore) {
            }
        }
    }

    @Override
    public Set<Timeout> stop() {
        if (Thread.currentThread() == workerThread) {
            throw new IllegalStateException(
                    GameWheelTimer.class.getSimpleName() +
                            ".stop() 不能在 TimerTask 内部调用");
        }

        if (!WORKER_STATE_UPDATER.compareAndSet(this, WORKER_STATE_STARTED, WORKER_STATE_SHUTDOWN)) {
            if (WORKER_STATE_UPDATER.getAndSet(this, WORKER_STATE_SHUTDOWN) != WORKER_STATE_SHUTDOWN) {
                INSTANCE_COUNTER.decrementAndGet();
                if (leak != null) {
                    boolean closed = leak.close(this);
                    assert closed;
                }
            }

            return Collections.emptySet();
        }

        try {
            //安全停止workerThread
            boolean interrupted = false;
            while (workerThread.isAlive()) {
                workerThread.interrupt();
                try {
                    workerThread.join(100);
                } catch (InterruptedException ignored) {
                    interrupted = true;
                }
            }

            if (interrupted) {
                Thread.currentThread().interrupt();
            }
        } finally {
            INSTANCE_COUNTER.decrementAndGet();
            if (leak != null) {
                boolean closed = leak.close(this);
                assert closed;
            }
        }
        return worker.unprocessedTimeouts();
    }

    // ========================== 新建定时任务 ==========================
    @Override
    public Timeout newTimeout(TimerTask task, long delay, TimeUnit unit) {
        if (task == null) {
            throw new NullPointerException("task");
        }
        if (unit == null) {
            throw new NullPointerException("unit");
        }

        long pendingTimeoutsCount = pendingTimeouts.incrementAndGet();

        if (maxPendingTimeouts > 0 && pendingTimeoutsCount > maxPendingTimeouts) {
            pendingTimeouts.decrementAndGet();
            throw new RejectedExecutionException("待处理任务数量 (" + pendingTimeoutsCount +
                    ") 超过最大允许值 (" + maxPendingTimeouts + ")");
        }

        start();

        // 计算任务截止时间并加入队列
        long deadline = System.nanoTime() + unit.toNanos(delay) - startTime;

        if (delay > 0 && deadline < 0) {
            deadline = Long.MAX_VALUE;
        }
        GameWheelTimer.HashedWheelTimeout timeout = new GameWheelTimer.HashedWheelTimeout(this, task, deadline);
        timeouts.add(timeout);
        return timeout;
    }

    public long pendingTimeouts() {
        return pendingTimeouts.get();
    }

    private static void reportTooManyInstances() {
        if (logger.isErrorEnabled()) {
            String resourceType = simpleClassName(GameWheelTimer.class);
            logger.error("创建了过多的 " + resourceType + " 实例。" +
                    resourceType + " 是共享资源，应在 JVM 内重用，" +
                    "确保只创建少量实例。");
        }
    }

    // ========================== Worker ==========================
    private final class Worker implements Runnable {
        private final Set<Timeout> unprocessedTimeouts = new HashSet<Timeout>();
        private long tick;

        @Override
        public void run() {
            startTime = System.nanoTime();
            if (startTime == 0) {
                startTime = 1;
            }
            startTimeInitialized.countDown();

            do {
                //sleep 直到下个tick
                final long deadline = waitForNextTick();
                if (deadline > 0) {
                    int idx = (int) (tick & mask);
                    processCancelledTasks();
                    GameWheelTimer.HashedWheelBucket bucket = wheel[idx];
                    transferTimeoutsToBuckets();
                    bucket.expireTimeouts(deadline);
                    tick++;
                }
            } while (WORKER_STATE_UPDATER.get(GameWheelTimer.this) == WORKER_STATE_STARTED);

            // 收集每个bucket的任务到 unprocessedTimeouts
            for (GameWheelTimer.HashedWheelBucket bucket: wheel) {
                bucket.clearTimeouts(unprocessedTimeouts);
            }
            // 收集timeous 队列的任务到 unprocessedTimeouts
            for (;;) {
                GameWheelTimer.HashedWheelTimeout timeout = timeouts.poll();
                if (timeout == null) {
                    break;
                }
                if (!timeout.isCancelled()) {
                    unprocessedTimeouts.add(timeout);
                }
            }
            //处理取消的任务 从Bucket里移除
            processCancelledTasks();
        }

        /**
         * 转移timeout 到时间轮
         */
        private void transferTimeoutsToBuckets() {
            for (int i = 0; i < 100000; i++) {
                GameWheelTimer.HashedWheelTimeout timeout = timeouts.poll();
                if (timeout == null) break;
                //取消的过滤掉
                if (timeout.state() == GameWheelTimer.HashedWheelTimeout.ST_CANCELLED) continue;

                long calculated = timeout.deadline / tickDuration;
                timeout.remainingRounds = (calculated - tick) / wheel.length;

                final long ticks = Math.max(calculated, tick);
                int stopIndex = (int) (ticks & mask);

                GameWheelTimer.HashedWheelBucket bucket = wheel[stopIndex];
                bucket.addTimeout(timeout);
            }
        }

        private void processCancelledTasks() {
            for (;;) {
                GameWheelTimer.HashedWheelTimeout timeout = cancelledTimeouts.poll();
                if (timeout == null) break;
                try {
                    timeout.remove();
                } catch (Throwable t) {
                    if (logger.isWarnEnabled()) {
                        logger.warn("处理取消任务时发生异常", t);
                    }
                }
            }
        }

        private long waitForNextTick() {
            long deadline = tickDuration * (tick + 1);

            for (;;) {
                final long currentTime = System.nanoTime() - startTime;
                long sleepTimeMs = (deadline - currentTime + 999999) / 1000000;

                if (sleepTimeMs <= 0) {
                    if (currentTime == Long.MIN_VALUE) {
                        return -Long.MAX_VALUE;
                    } else {
                        return currentTime;
                    }
                }

                if (PlatformDependent.isWindows()) {
                    sleepTimeMs = sleepTimeMs / 10 * 10;
                }

                try {
                    Thread.sleep(sleepTimeMs);
                } catch (InterruptedException ignored) {
                    if (WORKER_STATE_UPDATER.get(GameWheelTimer.this) == WORKER_STATE_SHUTDOWN) {
                        return Long.MIN_VALUE;
                    }
                }
            }
        }

        public Set<Timeout> unprocessedTimeouts() {
            return Collections.unmodifiableSet(unprocessedTimeouts);
        }
    }

    // ========================== HashedWheelTimeout ==========================
    private static final class HashedWheelTimeout implements Timeout {

        private static final int ST_INIT = 0;
        private static final int ST_CANCELLED = 1;
        private static final int ST_EXPIRED = 2;
        private static final AtomicIntegerFieldUpdater<GameWheelTimer.HashedWheelTimeout> STATE_UPDATER =
                AtomicIntegerFieldUpdater.newUpdater(GameWheelTimer.HashedWheelTimeout.class, "state");

        private final GameWheelTimer timer;
        private final TimerTask task;
        private final long deadline;
        /**
         * 原子更新这个值
         */
        private volatile int state = ST_INIT; // 状态：初始化、取消、过期
        long remainingRounds; // 剩余轮次
        GameWheelTimer.HashedWheelTimeout next;
        GameWheelTimer.HashedWheelTimeout prev;
        GameWheelTimer.HashedWheelBucket bucket;

        HashedWheelTimeout(GameWheelTimer timer, TimerTask task, long deadline) {
            this.timer = timer;
            this.task = task;
            this.deadline = deadline;
        }

        @Override
        public Timer timer() { return timer; }
        @Override
        public TimerTask task() { return task; }

        @Override
        public boolean cancel() {
            if (!compareAndSetState(ST_INIT, ST_CANCELLED)) return false;
            timer.cancelledTimeouts.add(this);
            return true;
        }

        void remove() {
            GameWheelTimer.HashedWheelBucket bucket = this.bucket;
            if (bucket != null) {
                bucket.remove(this);
            } else {
                timer.pendingTimeouts.decrementAndGet();
            }
        }

        public boolean compareAndSetState(int expected, int state) {
            return STATE_UPDATER.compareAndSet(this, expected, state);
        }

        public int state() { return state; }
        @Override
        public boolean isCancelled() { return state() == ST_CANCELLED; }
        @Override
        public boolean isExpired() { return state() == ST_EXPIRED; }

        public void expire() {
            if (!compareAndSetState(ST_INIT, ST_EXPIRED)) return;

            try {
                task.run(this);
            } catch (Throwable t) {
                if (logger.isWarnEnabled()) {
                    logger.warn("TimerTask 执行时发生异常", t);
                }
            }
        }

        @Override
        public String toString() {
            final long currentTime = System.nanoTime();
            long remaining = deadline - currentTime + timer.startTime;

            StringBuilder buf = new StringBuilder(192)
                    .append(simpleClassName(this))
                    .append('(')
                    .append("deadline: ");
            if (remaining > 0) {
                buf.append(remaining).append(" ns 后");
            } else if (remaining < 0) {
                buf.append(-remaining).append(" ns 前");
            } else {
                buf.append("现在");
            }

            if (isCancelled()) buf.append(", 已取消");

            return buf.append(", task: ").append(task()).append(')').toString();
        }
    }

    // ========================== HashedWheelBucket ==========================
    private static final class HashedWheelBucket {
        private GameWheelTimer.HashedWheelTimeout head;
        private GameWheelTimer.HashedWheelTimeout tail;

        public void addTimeout(GameWheelTimer.HashedWheelTimeout timeout) {
            assert timeout.bucket == null;
            timeout.bucket = this;
            if (head == null) {
                head = tail = timeout;
            } else {
                tail.next = timeout;
                timeout.prev = tail;
                tail = timeout;
            }
        }

        public void expireTimeouts(long deadline) {
            GameWheelTimer.HashedWheelTimeout timeout = head;
            while (timeout != null) {
                GameWheelTimer.HashedWheelTimeout next = timeout.next;
                if (timeout.remainingRounds <= 0) {
                    next = remove(timeout);
                    if (timeout.deadline <= deadline) {
                        timeout.expire();
                    } else {
                        throw new IllegalStateException(String.format(
                                "timeout.deadline (%d) > deadline (%d)", timeout.deadline, deadline));
                    }
                } else if (timeout.isCancelled()) {
                    next = remove(timeout);
                } else {
                    timeout.remainingRounds--;
                }
                timeout = next;
            }
        }

        public GameWheelTimer.HashedWheelTimeout remove(GameWheelTimer.HashedWheelTimeout timeout) {
            GameWheelTimer.HashedWheelTimeout next = timeout.next;
            if (timeout.prev != null) timeout.prev.next = next;
            if (timeout.next != null) timeout.next.prev = timeout.prev;

            if (timeout == head) {
                if (timeout == tail) {
                    tail = head = null;
                } else {
                    head = next;
                }
            } else if (timeout == tail) {
                tail = timeout.prev;
            }
            timeout.prev = timeout.next = null;
            timeout.bucket = null;
            timeout.timer.pendingTimeouts.decrementAndGet();
            return next;
        }

        public void clearTimeouts(Set<Timeout> set) {
            for (;;) {
                GameWheelTimer.HashedWheelTimeout timeout = pollTimeout();
                if (timeout == null) return;
                if (timeout.isExpired() || timeout.isCancelled()) continue;
                set.add(timeout);
            }
        }

        private GameWheelTimer.HashedWheelTimeout pollTimeout() {
            GameWheelTimer.HashedWheelTimeout head = this.head;
            if (head == null) return null;
            GameWheelTimer.HashedWheelTimeout next = head.next;
            if (next == null) {
                tail = this.head = null;
            } else {
                this.head = next;
                next.prev = null;
            }
            head.next = head.prev = null;
            head.bucket = null;
            return head;
        }
    }
}
